package com.atbeijing.gmall.realtime.utils;

import com.atbeijing.gmall.realtime.common.GmallConfig;
import com.atbeijing.gmall.realtime.common.TransientSink;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.lang.reflect.Field;
import java.sql.PreparedStatement;
import java.sql.SQLException;

/**
 * 操作Click house的工具类
 */
public class ClickHouseUtil {

    /**
     * 获取jdbc sink
     * @param sql
     * @param <T> 第一个T表示模板
     * @return
     */
    public static <T>SinkFunction<T> getJdbcSink(String sql){
        SinkFunction<T> sink = JdbcSink.<T>sink(
                sql,
                new JdbcStatementBuilder<T>() {
                    //insert into visitor_stats_1116 values(?,?,?,?,?,?,?,?,?,?,?,?)
                    //obj是流中的对象
                    //问号的个数代表向数据库插入数据的个数,如果obj中的字段被@TransientSink标记,就不能向数据库添加,?的个数就要减少
                    //obj中字段顺序要与数据库字段顺序一致
                    //数据库字段个数=问号个数=obj字段个数-被@TransientSink标记的字段个数
                    @Override
                    public void accept(PreparedStatement ps, T obj) throws SQLException {
                        //标识有多少个被@TransientSink标记的字段
                        int skipNum = 0;
                        //获取对象的所有字段,包括私有的
                        Field[] fields = obj.getClass().getDeclaredFields();
                        //获取每一个字段对象
                        for (int i = 0; i < fields.length; i++) {
                            Field field = fields[i];
                            //允许访问私有字段
                            field.setAccessible(true);
                            //判断当前字段是否被@TransientSink注解标记
                            TransientSink annotation = field.getAnnotation(TransientSink.class);
                            //如果有这个注解就跳过
                            if (annotation!=null){
                                skipNum++;
                                continue;
                            }
                            try {
                                //获取属性的值
                                Object fieldValue = field.get(obj);
                                //将属性的值赋值给问号占位符,sql的占位符是从1开始,所以要i+1
                                ps.setObject(i + 1 - skipNum,fieldValue);
                            } catch (IllegalAccessException e) {
                                e.printStackTrace();
                            }
                        }
                    }
                },
                //数据处理设置,每批处理5条数据
                new JdbcExecutionOptions.Builder().withBatchSize(5).build(),
                //连接数据库
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withDriverName(GmallConfig.CLICKHOUSE_DRIVER)
                        .withUrl(GmallConfig.CLICKHOUSE_URL)
                        .build()
        );
        return sink;
    }

}
